<?php

namespace common\helpers;
use Yii;
use yii\db\Exception;


/**
 * redis队列
 * Class Queue
 * @package common\helpers
 */
class Queue
{

    const PREFIX = 'redisQueue:';

    //延迟队列zet前缀
    const Q_Z_SET = 'QueueZset:';

    //延迟队列hashList前缀
    const Q_H_LIST = 'QueueHlist:';

    /**
     * @return yii\redis\Connection
     */
    public static function getRedis() {
        return Yii::$app->redis;
    }

    /**
     * 推入消息
     * @param string $key
     * @param string $message
     * @param int|null $time 消费时间（生效时间） 如1556179025 （2019/4/25 15:57:5 之后生效）
     * @param int $autoOverdueDay 延迟队列自动过期天数 （默认15）
     * @return mixed
     */

    public static function push($key, $message, $time = null, $autoOverdueDay = 15)
    {
        $key = self::PREFIX . $key;
        $redisClient = self::getRedis();
        $nowTime = time();
        if($time && $time > $nowTime) {
            //延迟队列
            $hashKey = md5(self::randStr(16) . ':' . $nowTime . ':'. $message);
            $zaddKey = self::Q_Z_SET . $key;
            $hashListKey = self::Q_H_LIST . $key;
            $overdueTime = $autoOverdueDay * 86400;
            //标记事务

            $redisClient->multi();
            $redisClient->zadd($zaddKey, $time, $hashKey);
            $redisClient->hset($hashListKey, $hashKey, $message);
            $redisClient->expire($zaddKey,$overdueTime);
            $redisClient->expire($hashListKey,$overdueTime);
            return $redisClient->exec() ? [
                'key' => $key,
                'hashKey' => $hashKey,
            ] : false;
        } else {
            //普通队列
            return $redisClient->rpush($key,$message) ? ['key'=>$key] : false;
        }
    }






    /**
     * 接收消息
     * @param $key
     * @param bool $syncWaitMessage
     * @param int $waitTime
     * @return |null
     * @throws \Throwable
     */
    public static function receive($key, $syncWaitMessage = false,$waitTime = 5)
    {
        $key = self::PREFIX . $key;
        $redisClient = self::getRedis();
        try {
            //同步等待消息
            if ($syncWaitMessage && self::getLock($key)) {
                self::syncWaitMessage($key);
                self::releaseLock($key);
            }
            $ret = $redisClient->blpop($key,$waitTime);
            return !empty($ret) ? $ret[1] : null;
        } catch (yii\redis\SocketException $ex){
            return null;
        } catch (\Throwable $e) {
            throw $e;
            //throw new \Exception('redis syncWaitMessage error');
        }
    }


    /**
     * 获取长度
     * @param $key
     * @return int
     */
    public static function getLen($key)
    {
        return self::getRedis()->llen(self::PREFIX . $key);
    }

    /**
     * 返回客户端
     * @return mixed
     */
    public static function getClient()
    {
        return self::getRedis();
    }


    /**
     * 同步等待消息
     * @param string $key 消息队列 key
     * @throws \yii\db\Exception
     */
    private static function syncWaitMessage($key)
    {
        $redisClient = self::getRedis();
        //到期的队列key集合
        $keys = $redisClient->zrangebyscore(self::Q_Z_SET . $key,0, time());
        if(empty($keys)) {
            return;
        }
        $messageData = $redisClient->executeCommand('HMGET', array_merge([self::Q_H_LIST . $key], $keys));
        $redisClient->multi();
        foreach($messageData as $i => $message) {
            $redisClient->rpush($key,$message);
            self::delayQueueDel($key,$keys[$i],false);
        }
        $redisClient->exec();
    }

    /**
     * 延迟队列删除
     * @param string $queueName 队列名
     * @param string $hashKey 延迟队列hashKey
     * @param bool $addPreFix 是否追加前缀
     */
    public static function delayQueueDel($queueName,$hashKey,$addPreFix = true)
    {

        if ($addPreFix) { $queueName = self::PREFIX . $queueName;}
        $redisClient = self::getRedis();
        $redisClient->hdel(self::Q_H_LIST . $queueName, $hashKey);
        $redisClient->zrem(self::Q_Z_SET . $queueName, $hashKey);
    }

    /**
     * 获取锁
     * @param string $key 锁键值
     * @param integer $wait_second 等待秒数
     * @return bool
     */
    protected static  function getLock($key, $maxLockTime=60)
    {
        $key = "queue:lock" . $key;
        $redisClient = self::getRedis();
        return $redisClient->set($key, 1, 'nx', 'ex' ,$maxLockTime);
    }

    /**
     * 解除锁
     * @param string $key 锁键值
     */
    protected static function releaseLock($key)
    {
        $key = "queue:lock" . $key;
        self::getRedis()->del($key);
    }

    private static function randStr($length) {
        $str = '';
        $strPol = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyz";
        $max = strlen($strPol) - 1;
        for ($i = 0; $i < $length; $i++) {
            $str .= $strPol[mt_rand(0, $max)];
        }
        return $str;
    }

}